package source;

import bean.baseTrademark;
import common.DBConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.concurrent.TimeUnit;

public class MyMysqlSource extends RichParallelSourceFunction<baseTrademark> {

    Connection connection = null;
    PreparedStatement preparedStatement = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(DBConfig.MYSQL_DRIVER);
        connection = DriverManager.getConnection(DBConfig.MYSQL_SERVER);
        String sql = "select * from base_trademark";
        preparedStatement = connection.prepareStatement(sql);
    }

    private boolean flag = true;

    @Override
    public void run(SourceContext<baseTrademark> sourceContext) throws Exception {
        while (flag){
            ResultSet resultSet = preparedStatement.executeQuery();
            while (resultSet.next()){
                int id = resultSet.getInt("id");
                String tmName = resultSet.getString("tmName");
                String logoUrl = resultSet.getString("logoUrl");
                sourceContext.collect(new baseTrademark(id,tmName,logoUrl));
            }
            TimeUnit.SECONDS.sleep(5);
        }
    }

    @Override
    public void cancel() {
        flag = false;
    }

    @Override
    public void close() throws Exception {
        if (connection!=null){
            connection.close();
        }
        if (preparedStatement!=null){
            preparedStatement.close();
        }
    }
}
